package org.example.service;

import cn.hutool.core.thread.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientRequestException;
import org.springframework.web.reactive.function.client.WebClientResponseException;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;

/**
 * AlertSenderService
 *
 * @author: wangchao
 * @Date: 2025/6/10 20:05
 * @since 7.0.0-RC2
 **/

@Slf4j
@Service
public class AlertSenderService {
    @Resource
    private WebClient webClient;
    @Resource
    private ScheduledExecutorService scheduledExecutorService;
    @Resource
    private NetworkConnectivityService networkConnectivityService;
    private int batchSize = 10;
    private long batchIntervalMs = 1000;
    private static final String url = "http://127.0.0.1:19001/agent/task/start";
    private static final String base_url = "http://127.0.0.1:19001";
    private BlockingQueue<Alert> alertQueue = new LinkedBlockingQueue<>();

    public void sendAlert(Alert alert) {
        alertQueue.add(alert);
    }

    @PostConstruct
    public void asyncSendAlert() {
        scheduledExecutorService.scheduleAtFixedRate(this::asyncSendAlertBatch, 0, batchIntervalMs,
            TimeUnit.MILLISECONDS);
        log.info("AlertSenderService initialized with batch size: {}, interval: {}ms", batchSize, batchIntervalMs);
    }

    private void asyncSendAlertBatch() {
        try {
            // 一次性从队列中获取最多batchSize条告警
            List<Alert> alertsToSend = new ArrayList<>(batchSize);
            alertQueue.drainTo(alertsToSend, batchSize);
            if (!alertsToSend.isEmpty()) {
                log.info("Sending batch of {} alerts", alertsToSend.size());
                sendAlerts(alertsToSend);
            }
        } catch (Exception e) {
            log.error("Error in alert batch processing", e);
        }
    }

    private void sendAlerts(List<Alert> alerts) {
        try {
            // 构建批量告警请求
            BatchAlertRequest batchRequest = new BatchAlertRequest(alerts);
            // 发送POST请求
            webClient.post()
                .uri(url)
                .bodyValue(batchRequest)
                .retrieve()
                .bodyToMono(Void.class)
                .timeout(Duration.ofSeconds(30))
                .doOnSuccess(v -> log.debug("Successfully sent {} alerts", alerts.size()))
                .doOnError(e -> log.error("Failed to send alerts", e))
                .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
                    .filter(throwable -> throwable instanceof WebClientRequestException
                        || throwable instanceof TimeoutException))
                .subscribe();
        } catch (Exception e) {
            // 发送失败时将告警放回队列
            log.error("Failed to send alerts, re-enqueuing", e);
            alerts.forEach(this::enqueueAlert);
        }
    }

    @PreDestroy
    public void shutdown() {
        log.info("Shutting down AlertSenderService");
        scheduledExecutorService.shutdown();
        try {
            if (!scheduledExecutorService.awaitTermination(5, TimeUnit.SECONDS)) {
                scheduledExecutorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            scheduledExecutorService.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    public void enqueueAlert(Alert alert) {
        if (alert != null) {
            alertQueue.offer(alert);
            log.debug("Alert enqueued: {}", alert.getId());
        }
    }

    public void connectingAgentServer() {
        boolean isConnected = false;
        while (!isConnected) {
            // 尝试连接到Agent服务器
            ConnectivityResult connectivityResult = networkConnectivityService.checkConnectivityWithRetry(base_url);
            isConnected = connectivityResult.isConnected();
            log.warn("connect to Agent server, retrying : {}", isConnected);
            ThreadUtil.safeSleep(1000);
        }
    }

    private static class BatchAlertRequest {
        private String clusterIp = "192.168.0.114";
        private String clusterPort = "25432";
        // alarm | trace
        private String messageType = "alarm";
        private final List<Alert> alerts;

        public BatchAlertRequest(List<Alert> alerts) {
            this.alerts = alerts;
        }

        public List<Alert> getAlerts() {
            return alerts;
        }
    }
}
